package com.chukun.flink.stream.operator.rich;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * @version 1.0.0
 * @description 富函数的基本使用
 * @createTime 2022年05月11日 23:17:00
 */
public class RichFunctionOperator extends RichFlatMapFunction<Long, Long> {

    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置作业的全局并行度2
        env.setParallelism(2);
        // 创建一个包含指定数字序列的新数据流
        DataStream<Long> dataStream = env.fromSequence(1, 100);

        DataStream<Long> mappedStream = dataStream.flatMap(new RichFunctionOperator())
                .name("inner-flatMap"); // 设置任务名称

        // 打印
        mappedStream.print();

        env.execute("RichFunctionOperator");
    }


    @Override
    public void flatMap(Long aLong, Collector<Long> collector) throws Exception {
        TimeUnit.SECONDS.sleep(1);
        collector.collect(aLong);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext rc = getRuntimeContext();
        String task = rc.getTaskName();
        String subTaskName = rc.getTaskNameWithSubtasks();
        int index = rc.getIndexOfThisSubtask();
        int parallel = rc.getNumberOfParallelSubtasks();
        int attemptNumber = rc.getAttemptNumber();
        System.out.println("调用open方法,初始化资源信息");
        System.out.printf("调用open方法,任务名称 : %s, 子任务名称 : %s,并行子任务标识 : %d, 当前任务的总并行度: %d", task, subTaskName, index, parallel);
        System.out.println();
        System.out.printf("调用open方法,该任务失败重启的次数: %d", attemptNumber);
        System.out.println();
    }

    @Override
    public void close() throws Exception {
        System.out.println("调用close方法......");
    }
}
